#include "StdAfx.h"
#include "JoinOperator.h"
#include "LineageSynopsis.h"
#include "QueueEntity.h"
#include "DocumentIdentifierGenerator.h"
#include "RelationSynopsis.h"
#include "ElementIterator.h"
JoinOperator::JoinOperator(void)
{
	int lineageNumber = 2;
	this->lineageSynopsis.reset(new LineageSynopsis(lineageNumber));
	this->leftRelationSynopsis.reset(new RelationSynopsis());
	this->rightRelationSynopsis.reset(new RelationSynopsis());
	this->lastLeftTimestamp = 0;
	this->lastRightTimestamp = 0;
	this->leftElementIterator.reset(new ElementIterator(this->leftRelationSynopsis));
	this->rightElementIterator.reset(new ElementIterator(this->rightRelationSynopsis));

}


JoinOperator::~JoinOperator(void)
{
}

void JoinOperator::setLeftJoinAttribute(std::string leftJoinAttribute)
{
	this->leftJoinAttribute = leftJoinAttribute;
}
void JoinOperator::setRightJoinAttribute(std::string rightJoinAttribute)
{
	this->rightJoinAttribute = rightJoinAttribute;
}
void JoinOperator::execution(int executionTimes)
{
#ifdef DEBUG
	std::cout<<"===================operator begin================="<<std::endl;
	std::cout<<"operatorid : "<<this->getId()<<std::endl;
#endif
	assert(this->getInputQueueList().size()==2);
	assert(this->getOutputQueueList().size()==1);
	boost::shared_ptr<QueueEntity>leftInputQueue = this->getInputQueueList().front();
	boost::shared_ptr<QueueEntity>rightInputQueue = this->getInputQueueList().back();
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Element leftElement;
	Element rightElement;
	Timestamp minimumLeftQueueLastTimestamp;
	Timestamp minimumRightQueueLastTimestamp;
	minimumLeftQueueLastTimestamp = lastLeftTimestamp;
	minimumRightQueueLastTimestamp = lastRightTimestamp;
	
	if(this->getOperatorState()==BLOCKED)
	{
		dealWithBlock();
		if(this->getOperatorState()==BLOCKED)
		{
#ifdef DEBUG
			std::cout<<"operator blocked"<<std::endl;
			std::cout<<"===================operator over================="<<std::endl;
#endif
			return;
		}
	}

	for(int i = 0 ;i<executionTimes ; i++)
	{
		if(outputQueue->isFull())
		{
			break;
		}
		if(!leftInputQueue->isEmpty())
		{
			leftInputQueue->peek(leftElement);
			minimumLeftQueueLastTimestamp = leftElement.timestamp;
		}
		if(!rightInputQueue->isEmpty())
		{
			rightInputQueue->peek(rightElement);
			minimumRightQueueLastTimestamp = rightElement.timestamp;
		}

		
		if(minimumLeftQueueLastTimestamp<minimumRightQueueLastTimestamp)
		{
			if(leftInputQueue->isEmpty())
			{
				break;
			}
			leftInputQueue->dequeue();
			lastLeftTimestamp = leftElement.timestamp;
			if(leftElement.mark == PLUS_MARK)
			{
				processLeftPlus(leftElement);
			}
			else if(leftElement.mark == MINUS_MARK)
			{
				processLeftMinus(leftElement);
			}
		}
		else if (minimumLeftQueueLastTimestamp > minimumRightQueueLastTimestamp)
		{
			if(rightInputQueue->isEmpty())
			{
				break;
			}
			rightInputQueue -> dequeue();
			lastRightTimestamp = rightElement.timestamp;
			if(rightElement.mark==PLUS_MARK)
			{
				processRightPlus(rightElement);
			}
			else if(rightElement.mark == MINUS_MARK)
			{
				processRightMinus(rightElement);
			}
		}
		else if(!leftInputQueue->isEmpty())//minimumLeftQueueLastTimestamp == minimumRightQueueLastTimestamp
		{

			leftInputQueue->dequeue();
			lastLeftTimestamp = leftElement.timestamp;
			if(leftElement.mark==PLUS_MARK)
			{
				processLeftPlus(leftElement);
			}
			else if(leftElement.mark == MINUS_MARK)
			{
				processLeftMinus(leftElement);
			}
		}
		else if(!rightInputQueue->isEmpty())
		{
			rightInputQueue -> dequeue();
			lastRightTimestamp = rightElement.timestamp;
			if(rightElement.mark==PLUS_MARK)
			{
				processRightPlus(rightElement);
			}
			else if(rightElement.mark == MINUS_MARK)
			{
				processRightMinus(rightElement);
			}
		}
	}
#ifdef DEBUG
	std::cout<<"===================operator over================="<<std::endl;
#endif
}
void JoinOperator::processLeftPlus (Element& leftElement)
{

	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Document& leftDocument = leftElement.document;
	if(this->leftRelationSynopsis->isFull())
	{
		throw runtime_error("left synopsis is full");
	}
	this->leftRelationSynopsis->insertElement(leftElement);

	
	this->rightElementIterator->initial();
	Element rightElement;
	while((!outputQueue->isFull())&&this->rightElementIterator->getNext(rightElement))
	{
		
		Document& rightDocument = rightElement.document;
		Element newElement;
		Document newDocument;
		DocumentBuilder newDocumentBuilder;
		//std::cout<<"left document : "<<leftDocument<<std::endl;
		//std::cout<<"right document : "<<rightDocument<<std::endl;
		if(!(leftDocument.getField(this->leftJoinAttribute)==rightDocument.getField(this->rightJoinAttribute)))
		{
			continue;
		}
		

		//std::set<std::string> leftDocumentFieldSet;
		//std::set<std::string> rightDocumentFieldSet;
		//leftDocument.getFieldNames(leftDocumentFieldSet);
		//rightDocument.getFieldNames(rightDocumentFieldSet);
		//rightDocumentFieldSet.erase(rightDocumentFieldSet.find(rightJoinAttribute));
		//std::cout<<leftDocument<<std::endl;
		//std::cout<<rightDocument<<std::endl;
		newDocumentBuilder.appendElements(leftDocument);
		newDocumentBuilder.appendElements(rightDocument.removeField(this->rightJoinAttribute));
		newDocument = newDocumentBuilder.obj();
		//std::cout<<newDocument<<std::endl;
		//std::cout<<rightDocument<<std::endl;
		//	newDocument.addFields(leftDocument,leftDocumentFieldSet);
		
		//std::cout<<newDocument<<std::endl;
		newElement.id = DocumentIdentifierGenerator::generateNewDocumentIdentifier();
		newElement.document = newDocument;
		newElement.mark = PLUS_MARK;
		newElement.timestamp = leftElement.timestamp;

		//insert into lineage synopsis
		if(this->lineageSynopsis->isFull())
		{
			throw runtime_error("lineage synopsis is full");
		}
		Lineage lineage;
		lineage.lineageNumber = 2;
		lineage.lineageDocumentId[0] = leftElement.id;
		lineage.lineageDocumentId[1] = rightElement.id;
		this->lineageSynopsis->insertLineage(lineage,newElement);

		//generate new output to the output queue
		if(outputQueue->isFull())
		{
			throw std::runtime_error("output queue is full");
		}
		outputQueue->enqueue(newElement);
	}
	if(outputQueue->isFull())// we should block the operator
	{
		this->setOperatorState(BLOCKED);
		blockElement = leftElement;
		this->joinBlockState = LEFT_PLUS_BLOCK;
	}

	

}
void JoinOperator::processLeftMinus (Element& leftElement)
{
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Document& leftDocument = leftElement.document;
	this->leftRelationSynopsis->deleteElement(leftElement);
	this->rightElementIterator->initial();
	Element rightElement;
	while((!outputQueue->isFull())&&this->rightElementIterator->getNext(rightElement))
	{
	
		Document& rightDocument = rightElement.document;
		//std::cout<<"left document : "<<leftDocument<<std::endl;
		//std::cout<<"right document : "<<rightDocument<<std::endl;
		//if satisfied join condition
		if(!(leftDocument.getField(this->leftJoinAttribute)==rightDocument.getField(this->rightJoinAttribute)))
		{
			continue;
		}
		Lineage lineage;
		lineage.lineageNumber = 2;
		lineage.lineageDocumentId[0] = leftElement.id;
		lineage.lineageDocumentId[1] = rightElement.id;
		Element elementInSynopsis;
		this->lineageSynopsis->getAndDeleteElement(lineage,elementInSynopsis);

		//generate output element
		Element outputElement;
		outputElement.mark = MINUS_MARK;
		outputElement.id = elementInSynopsis.id;
		outputElement.timestamp = leftElement.timestamp;
		outputElement.document = elementInSynopsis.document;

		//output element
		if(outputQueue->isFull())
		{
			throw std::runtime_error("output queue is full");
		}
		outputQueue->enqueue(outputElement);
	}
	if(outputQueue->isFull())// we should block the operator
	{
		this->setOperatorState(BLOCKED);
		blockElement = leftElement;
		this->joinBlockState = LEFT_MINUS_BLOCK;
	}
}
void JoinOperator::processRightPlus (Element& rightElement)
{
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Document rightDocument = rightElement.document;
	if(this->rightRelationSynopsis->isFull())
	{
		throw runtime_error("right synopsis is full");
	}
	this->rightRelationSynopsis->insertElement(rightElement);

	this->leftElementIterator->initial();
	Element leftElement;
	while((!outputQueue->isFull())&&this->leftElementIterator->getNext(leftElement))
	{
	
		Document leftDocument = leftElement.document;
		Element newElement;
		Document newDocument;
		DocumentBuilder newDocumentBuilder;
		//std::cout<<"left document : "<<leftDocument<<std::endl;
		//std::cout<<"right document : "<<rightDocument<<std::endl;
		if(!(leftDocument.getField(this->leftJoinAttribute)==rightDocument.getField(this->rightJoinAttribute)))
		{
			continue;
		}

		//std::set<std::string> leftDocumentFieldSet;
		//std::set<std::string> rightDocumentFieldSet;
		//leftDocument.getFieldNames(leftDocumentFieldSet);
		//rightDocument.getFieldNames(rightDocumentFieldSet);
		//	rightDocumentFieldSet.erase(rightDocumentFieldSet.find(rightJoinAttribute));

		newDocumentBuilder.appendElements(leftDocument);
		newDocumentBuilder.appendElements(rightDocument.removeField(this->rightJoinAttribute));
		newDocument = newDocumentBuilder.obj();
		//	newDocument.addFields(leftDocument,leftDocumentFieldSet);
		//std::cout<<leftDocument<<std::endl;
		//std::cout<<rightDocument<<std::endl;
		//std::cout<<newDocument<<std::endl;
		//	newDocument.addFields(rightDocument,rightDocumentFieldSet);
		newElement.id = DocumentIdentifierGenerator::generateNewDocumentIdentifier();
		newElement.document = newDocument;
		newElement.mark = PLUS_MARK;
		newElement.timestamp = rightElement.timestamp;

		//insert into lineage synopsis
		if(this->lineageSynopsis->isFull())
		{
			throw runtime_error("lineage synopsis is full");
		}
		Lineage lineage;
		lineage.lineageNumber = 2;
		lineage.lineageDocumentId[0] = leftElement.id;
		lineage.lineageDocumentId[1] = rightElement.id;
		this->lineageSynopsis->insertLineage(lineage,newElement);

		//generate new output to the output queue
		if(outputQueue->isFull())
		{
			throw std::runtime_error("output queue is full");
		}
		outputQueue->enqueue(newElement);
	}
	if(outputQueue->isFull())// we should block the operator
	{
		this->setOperatorState(BLOCKED);
		blockElement = rightElement;
		this->joinBlockState = RIGHT_PLUS_BLOCK;
	}
}
void JoinOperator::processRightMinus (Element& rightElement)
{
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Document& rightDocument = rightElement.document;
	this->rightRelationSynopsis->deleteElement(rightElement);
	this->leftElementIterator->initial();
	Element leftElement;
	while((!outputQueue->isFull())&&this->leftElementIterator->getNext(leftElement))
	{
	
		Document& leftDocument = leftElement.document;
		//if satisfied join condition
		//std::cout<<"left document : "<<leftDocument<<std::endl;
		//std::cout<<"right document : "<<rightDocument<<std::endl;
		if(!(leftDocument.getField(this->leftJoinAttribute)==rightDocument.getField(this->rightJoinAttribute)))
		{
			continue;
		}
		Lineage lineage;
		lineage.lineageNumber = 2;
		lineage.lineageDocumentId[0] = leftElement.id;
		lineage.lineageDocumentId[1] = rightElement.id;
		Element elementInSynopsis;
		this->lineageSynopsis->getAndDeleteElement(lineage,elementInSynopsis);

		//generate output element
		Element outputElement;
		outputElement.mark = MINUS_MARK;
		outputElement.id = elementInSynopsis.id;
		outputElement.timestamp = rightElement.timestamp;
		outputElement.document = elementInSynopsis.document;

		//output element
		if(outputQueue->isFull())
		{
			throw std::runtime_error("output queue is full");
		}
		outputQueue->enqueue(outputElement);
	}
	if(outputQueue->isFull())// we should block the operator
	{
		this->setOperatorState(BLOCKED);
		blockElement = rightElement;
		this->joinBlockState = RIGHT_MINUS_BLOCK;
	}
}
void JoinOperator::dealWithBlock(void)
{
#ifdef DEBUG
	std::cout<<"**********deal with block begin************"<<std::endl;
#endif
	assert(this->getOperatorState()==BLOCKED);
	this->setOperatorState(NORMAL);
	if(this->joinBlockState == LEFT_PLUS_BLOCK)
	{
		leftPlusBlock();
	}
	else if(this->joinBlockState == LEFT_MINUS_BLOCK)
	{
		leftMinusBlock();
	}
	else if(this->joinBlockState == RIGHT_PLUS_BLOCK)
	{
		rightPlusBlock();
	}
	else if(this->joinBlockState == RIGHT_MINUS_BLOCK)
	{
		rightMinusBlock();
	}
#ifdef DEBUG
	std::cout<<"**********deal with block over************"<<std::endl;
#endif
}
void JoinOperator::leftPlusBlock(void)
{
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Element& leftElement = this->blockElement;
	Document& leftDocument = leftElement.document;
	Element rightElement;
	while((!outputQueue->isFull())&&this->rightElementIterator->getNext(rightElement))
	{

		Document rightDocument = rightElement.document;
		Element newElement;
		Document newDocument;
		DocumentBuilder newDocumentBuilder;
		//std::cout<<"left document : "<<leftDocument<<std::endl;
		//std::cout<<"right document : "<<rightDocument<<std::endl;
		if(!(leftDocument.getField(this->leftJoinAttribute)==rightDocument.getField(this->rightJoinAttribute)))
		{
			continue;
		}


		std::set<std::string> leftDocumentFieldSet;
		std::set<std::string> rightDocumentFieldSet;
		leftDocument.getFieldNames(leftDocumentFieldSet);
		rightDocument.getFieldNames(rightDocumentFieldSet);
		//rightDocumentFieldSet.erase(rightDocumentFieldSet.find(rightJoinAttribute));
		newDocumentBuilder.appendElementsUnique(leftDocument);
		newDocumentBuilder.appendElementsUnique(rightDocument);
		newDocument = newDocumentBuilder.obj();
		//	newDocument.addFields(leftDocument,leftDocumentFieldSet);

		//std::cout<<newDocument<<std::endl;
		newElement.id = DocumentIdentifierGenerator::generateNewDocumentIdentifier();
		newElement.document = newDocument;
		newElement.mark = PLUS_MARK;
		newElement.timestamp = leftElement.timestamp;

		//insert into lineage synopsis
		if(this->lineageSynopsis->isFull())
		{
			throw runtime_error("lineage synopsis is full");
		}
		Lineage lineage;
		lineage.lineageNumber = 2;
		lineage.lineageDocumentId[0] = leftElement.id;
		lineage.lineageDocumentId[1] = rightElement.id;
		this->lineageSynopsis->insertLineage(lineage,newElement);

		//generate new output to the output queue
		if(outputQueue->isFull())
		{
			throw std::runtime_error("output queue is full");
		}
		outputQueue->enqueue(newElement);
	}
	if(outputQueue->isFull())// we should block the operator
	{
		this->setOperatorState(BLOCKED);
		blockElement = leftElement;
		this->joinBlockState = LEFT_PLUS_BLOCK;
	}
}
void JoinOperator::leftMinusBlock(void)
{
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Element& leftElement = blockElement;
	Document& leftDocument = leftElement.document;

	Element rightElement;
	while((!outputQueue->isFull())&&this->rightElementIterator->getNext(rightElement))
	{

		Document& rightDocument = rightElement.document;
		//std::cout<<"left document : "<<leftDocument<<std::endl;
		//std::cout<<"right document : "<<rightDocument<<std::endl;
		//if satisfied join condition
		if(!(leftDocument.getField(this->leftJoinAttribute)==rightDocument.getField(this->rightJoinAttribute)))
		{
			continue;
		}
		Lineage lineage;
		lineage.lineageNumber = 2;
		lineage.lineageDocumentId[0] = leftElement.id;
		lineage.lineageDocumentId[1] = rightElement.id;
		Element elementInSynopsis;
		this->lineageSynopsis->getAndDeleteElement(lineage,elementInSynopsis);

		//generate output element
		Element outputElement;
		outputElement.mark = MINUS_MARK;
		outputElement.id = elementInSynopsis.id;
		outputElement.timestamp = leftElement.timestamp;
		outputElement.document = elementInSynopsis.document;

		//output element
		if(outputQueue->isFull())
		{
			throw std::runtime_error("output queue is full");
		}
		outputQueue->enqueue(outputElement);
	}
	if(outputQueue->isFull())// we should block the operator
	{
		this->setOperatorState(BLOCKED);
		blockElement = leftElement;
		this->joinBlockState = LEFT_MINUS_BLOCK;
	}
}
void JoinOperator::rightPlusBlock(void)
{
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Element& rightElement = blockElement;
	Document& rightDocument = rightElement.document;
	Element leftElement;
	while((!outputQueue->isFull())&&this->leftElementIterator->getNext(leftElement))
	{

		Document leftDocument = leftElement.document;
		Element newElement;
		Document newDocument;
		DocumentBuilder newDocumentBuilder;
		//std::cout<<"left document : "<<leftDocument<<std::endl;
		//std::cout<<"right document : "<<rightDocument<<std::endl;
		if(!(leftDocument.getField(this->leftJoinAttribute)==rightDocument.getField(this->rightJoinAttribute)))
		{
			continue;
		}

		std::set<std::string> leftDocumentFieldSet;
		std::set<std::string> rightDocumentFieldSet;
		leftDocument.getFieldNames(leftDocumentFieldSet);
		rightDocument.getFieldNames(rightDocumentFieldSet);
		//	rightDocumentFieldSet.erase(rightDocumentFieldSet.find(rightJoinAttribute));

		newDocumentBuilder.appendElementsUnique(leftDocument);
		newDocumentBuilder.appendElementsUnique(rightDocument);
		newDocument = newDocumentBuilder.obj();
		//	newDocument.addFields(leftDocument,leftDocumentFieldSet);
		//std::cout<<leftDocument<<std::endl;
		//std::cout<<rightDocument<<std::endl;
		//std::cout<<newDocument<<std::endl;
		//	newDocument.addFields(rightDocument,rightDocumentFieldSet);
		newElement.id = DocumentIdentifierGenerator::generateNewDocumentIdentifier();
		newElement.document = newDocument;
		newElement.mark = PLUS_MARK;
		newElement.timestamp = rightElement.timestamp;

		//insert into lineage synopsis
		if(this->lineageSynopsis->isFull())
		{
			throw runtime_error("lineage synopsis is full");
		}
		Lineage lineage;
		lineage.lineageNumber = 2;
		lineage.lineageDocumentId[0] = leftElement.id;
		lineage.lineageDocumentId[1] = rightElement.id;
		this->lineageSynopsis->insertLineage(lineage,newElement);

		//generate new output to the output queue
		if(outputQueue->isFull())
		{
			throw std::runtime_error("output queue is full");
		}
		outputQueue->enqueue(newElement);
	}
	if(outputQueue->isFull())// we should block the operator
	{
		this->setOperatorState(BLOCKED);
		blockElement = rightElement;
		this->joinBlockState = RIGHT_PLUS_BLOCK;
	}
}
void JoinOperator::rightMinusBlock(void)
{
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Element& rightElement = blockElement;
	Document& rightDocument = rightElement.document;
	Element leftElement;
	while((!outputQueue->isFull())&&this->leftElementIterator->getNext(leftElement))
	{

		Document& leftDocument = leftElement.document;
		//if satisfied join condition
		//std::cout<<"left document : "<<leftDocument<<std::endl;
		//std::cout<<"right document : "<<rightDocument<<std::endl;
		if(!(leftDocument.getField(this->leftJoinAttribute)==rightDocument.getField(this->rightJoinAttribute)))
		{
			continue;
		}
		Lineage lineage;
		lineage.lineageNumber = 2;
		lineage.lineageDocumentId[0] = leftElement.id;
		lineage.lineageDocumentId[1] = rightElement.id;
		Element elementInSynopsis;
		this->lineageSynopsis->getAndDeleteElement(lineage,elementInSynopsis);

		//generate output element
		Element outputElement;
		outputElement.mark = MINUS_MARK;
		outputElement.id = elementInSynopsis.id;
		outputElement.timestamp = rightElement.timestamp;
		outputElement.document = elementInSynopsis.document;

		//output element
		if(outputQueue->isFull())
		{
			throw std::runtime_error("output queue is full");
		}
		outputQueue->enqueue(outputElement);
	}
	if(outputQueue->isFull())// we should block the operator
	{
		this->setOperatorState(BLOCKED);
		blockElement = rightElement;
		this->joinBlockState = RIGHT_MINUS_BLOCK;
	}
}